package com.we.risk.plist;

import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.streaming.util.serialization.KeyedSerializationSchema;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.we.risk.plist.utils.PlistTools;
import com.we.utils.CommonTimeTools;
import com.we.utils.ParamCheck;
import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;

import static com.we.risk.plist.utils.PlistTools.*;

public class PlstLocalDataCmp {
    public static final String RELEASEPROP = "risk/plist/adm_plist_kfk_self_test.properties";
    public static final String PROPSETS = "risk/plist/plist20230406.properties";
    public static Logger LOG = LoggerFactory.getLogger(PlstLocalDataCmp.class);

    //    public static final String KFKBROKERHOSTBAK =
    // "10.10.13.22:9092,10.10.13.23:9092,10.10.13.24:9092";
    //    public static final String KFKBAKTOPIC = "kafka.topic.data.record";
    public static final String KFKBROKERHOSTBAK =
            "10.10.15.25:9092,10.10.15.26:9092,10.10.15.27:9092";
    public static final String KFKBAKTOPIC = "AAA_risk_plist_source_test2";
    public static final String KFKBAKTOPICGRPID = "AdmPlist_grp";

    public static final String APPTAGDATA = "hdfs:///data/dim/app_data_tag/apk_tag_data";
    public static final String APPTAGFILENAME = "app_apk_tag";
    public static final String HFQAPPNAME = "好分期";

    public static final String SPLITWORD = "#";
    public static final String DATEFORMAT = "yyyy-MM-dd";
    public static final String MSGTYPE = "HFQ_UPLOAD_PLIST_RECORD";
    public static final String INPUTKEYBY = "userKey";
    public static final String OUTPUTKEY = "user_key";
    public static final String HFQINSTIME = "hfq_install_time";
    public static final String HTTPTIMESTAMP = "httpTimestamp";
    public static final String CAPTIME = "capture_time";
    public static final String LABELOUTPUTIME = "label_output_time";
    public static final String JOBID = "jobID";
    public static final String APPLIST = "appList";
    public static final String LASTOPENTIME = "lastOpenTime";
    public static final String LASTUPDATETIME = "lastUpdateTime";
    public static final String LASTISTIME = "lastInstallTime";
    public static final String APPNAME = "appName";
    public static final String APKNAME = "apkName";
    public static final String APPAPK = "app_apk";
    public static final String PACKAGENAME = "packageName";
    public static final String AFTHFQ = "is_aft_hfq";
    public static final String BFHFQ = "is_bf_hfq";
    public static final long INVAILDTIME = 315417600000L;
    public static final String APPAPKTAGFILE =
            "/Users/jk/Downloads/AAAAAAARzSz/UserLabelPlatform/14.4/apk_tag_data"; // 0743a00b8d06947da3bc36fcb1ef850d
    public static final String INPUTJSON =
            "/Users/jk/Downloads/风控标签需求/实时需求/plist/compare/1110/70cdcb0b15be86dcd4f1acef861b0a73.json";
    public static Long hfqInstallTime = null;
    public static HashMap<String, ArrayList<String>> appTagMap = new HashMap<>();
    public static HashMap<String, ArrayList<String>> apkTagMap = new HashMap<>();

    public static void main(String[] args) throws IOException {
        File inputFile = new File(INPUTJSON);

        String input = FileUtils.readFileToString(inputFile);
        JSONObject clearJson = cleanData(input);
        JSONObject jsonObject = ProcessData(clearJson);
        System.out.println(jsonObject.size());
        System.out.println(jsonObject.toString());
    }

    private static JSONObject cleanData(String input) {
        JSONObject outJson = null;
        try {
            if (input.contains(MSGTYPE)) {
                JSONObject inputJson = JSONObject.parseObject(input);
                if (inputJson != null && inputJson.getString("type").equals(MSGTYPE)) {
                    JSONObject dataJson = inputJson.getJSONObject("dataJson");
                    ParamCheck.isNull(dataJson, "dataJson");
                    JSONObject job = dataJson.getJSONObject("job");
                    String jobID = job.getString(JOBID);
                    ParamCheck.isNull(dataJson, "jobID");

                    String userKey = dataJson.getString(INPUTKEYBY);
                    ParamCheck.isNull(userKey, INPUTKEYBY);

                    JSONObject data = dataJson.getJSONObject("data");
                    JSONObject baseData = data.getJSONObject("baseData");
                    ParamCheck.isNull(baseData, "baseData");
                    Long captureTime = baseData.getLong(HTTPTIMESTAMP);
                    ParamCheck.isNull(captureTime, "captureTime");

                    Integer isAuthorized = baseData.getInteger("isAuthorized");
                    if (isAuthorized != null && isAuthorized != 3) {
                        outJson = appListFilter(data);
                        outJson.put(OUTPUTKEY, userKey);
                        outJson.put(CAPTIME, captureTime);
                        outJson.put(JOBID, jobID);
                        return outJson;
                    }
                }
            }
        } catch (Exception e) {
            LOG.error(e.toString());
            LOG.error("flatMap1 error input: " + input);
        }
        return outJson;
    }

    private static JSONObject appListFilter(JSONObject input) {
        JSONObject res = new JSONObject();
        JSONArray jsonArray = input.getJSONArray(APPLIST);
        int size = jsonArray.size();
        JSONArray resArray = new JSONArray();
        for (int i = 0; i < size; i++) {
            JSONObject callDetail = (JSONObject) jsonArray.get(i);
            String installTime = callDetail.getString(LASTOPENTIME);
            if (installTime != null && !installTime.equals("")) {
                String apkName = callDetail.getString(PACKAGENAME);
                callDetail.remove(PACKAGENAME);
                callDetail.put(APKNAME, apkName);
                //                    String appName = callDetail.getString(APPNAME);
                //                    String app_apk = (appName == null ? "" : appName) + "|" +
                // (apkName == null ? "" : apkName);
                //                    callDetail.put(APPAPK,app_apk);
                String appName = callDetail.getString(APPNAME);
                if (appName != null && appName.equals(HFQAPPNAME)) {
                    //                    hfqInstallTime = callDetail.getLong(installTime);
                    res.put(HFQINSTIME, installTime);
                }
                callDetail.remove(LASTOPENTIME);
                callDetail.put(LASTISTIME, installTime);
                resArray.add(callDetail);
            }
        }
        res.put(APPLIST, resArray);
        return res;
    }

    private static JSONObject ProcessData(JSONObject inputJson) {
        JSONObject resJson = null;

        try {
            SimpleDateFormat dateFormat = new SimpleDateFormat(DATEFORMAT);

            initTagMap();

            HashMap<String, AppCategoryBasicInfo> cateMap = initMap();
            AppCategoryBasicInfo totInfo = cateMap.get(TAGTOT);

            //            JSONObject inputJson = JSONObject.parseObject(input);
            Long cap_time = inputJson.getLong(CAPTIME);
            //                int cap_hour = CommonTimeTools.getHourOfTime(cap_time);
            String user_key = inputJson.getString(OUTPUTKEY);
            String job_id = inputJson.getString(JOBID);
            Long hfqInstTime = inputJson.getLong(HFQINSTIME);
            JSONArray app_list = inputJson.getJSONArray(APPLIST);
            int size = app_list.size();
            for (int i = 0; i < size; i++) {
                JSONObject appDetail = (JSONObject) app_list.get(i);
                String apk_name = appDetail.getString(APKNAME);
                String app_name = appDetail.getString(APPNAME);
                String app_apk =
                        (app_name == null ? "" : app_name)
                                + "|"
                                + (apk_name == null ? "" : apk_name);

                byte loanTag = PlistTools.getLoanTag(app_name, apk_name);
                byte bankTag = PlistTools.getBankTag(app_name);
                byte betTag = PlistTools.getBetTag(app_name, apk_name);
                Long install_time = appDetail.getLong(LASTISTIME);
                String inst_hfq_tag = null;
                //                if (hfqInstTime == null) {
                //                    inst_hfq_tag = "NONE";
                //                } else {
                //                    inst_hfq_tag = (install_time > hfqInstTime ? AFTHFQ : BFHFQ);
                //                }

                if (hfqInstTime != null) {
                    inst_hfq_tag = (install_time > hfqInstTime ? AFTHFQ : BFHFQ);
                } else {
                    inst_hfq_tag = "NONE";
                }

                if (bankTag == 1 && inst_hfq_tag.equals(BFHFQ)) {
                    LOG.error("Bank loanTag app_apk:" + app_apk);
                }

                String inst_h_part = getHourPart(install_time);

                Long update_time = appDetail.getLong(LASTUPDATETIME);
                String updt_h_part = getHourPart(update_time);

                ArrayList<String> appTags = getAppTag(app_name, apk_name, appTagMap, apkTagMap);
                //                String categoryTag = (String) appTag;
                //                for (String appTag: appTags) {
                //                    if(appTag.equals(TAGJRLC)) {
                //                        LOG.error("Debug TAGJRLC appApk: " + app_apk + "
                // updt_h_part: " + updt_h_part);
                //                    }
                //                }
                /** INVAILDTIME 返回NULL */
                Integer inst_day = dateBetween(cap_time, install_time, dateFormat);
                Integer inst_mth = monthBetween(cap_time, install_time);
                Integer updt_day = dateBetween(cap_time, update_time, dateFormat);
                Integer updt_mth = monthBetween(cap_time, update_time);

                Integer inst_day_cmp_hfq = dateBetween(install_time, hfqInstTime, dateFormat);
                //                if(updt_mth >= 1 && updt_mth <= 6 && bankTag == 1) {
                //                if(app_apk.equals("中国联通|com.sinovatech.unicom.ui") ||
                // app_apk.equals("HMS Core|com.huawei.hwid")
                //                || app_apk.equals("QQ同步助手|com.tencent.qqpim") ||
                // app_apk.equals("Breeno 指令|com.coloros.shortcuts")
                //                || app_apk.equals("QQ浏览器|com.tencent.mtt") ||
                // app_apk.equals("天气|com.coloros.weather2")
                //                || app_apk.equals("指南针|com.coloros.compass2") ||
                // app_apk.equals("计算器|com.coloros.calculator")
                //                || app_apk.equals("小X分身|com.bly.dkplat") ||
                // app_apk.equals("录音|com.coloros.soundrecorder")){
                //                if (app_apk.equals("指南针|com.coloros.compass2")) {
                //                    LOG.error("Bank updt_m_cnt app_apk: " + app_apk + " updt_day:
                // " + updt_day + " inst_day:"+ inst_day +" inst_mth: " + inst_mth);
                //                }
                //                if(bankTag == 1) {
                //                    LOG.error("Bank bankTag app_apk: " + app_apk + " updt_mth: " +
                // updt_mth  +" inst_mth: " + inst_mth);
                //                }

                AppApkObject appApkObj =
                        new AppApkObject(
                                app_apk,
                                inst_h_part,
                                updt_h_part,
                                inst_day,
                                inst_day_cmp_hfq,
                                updt_day,
                                inst_mth,
                                updt_mth,
                                appTags,
                                inst_hfq_tag);

                //                LOG.error("Debug appApkObj: " + appApkObj.toString());

                InsertApkToCategory(totInfo, appApkObj);

                if (loanTag == 1) {
                    InsertApkToCategory(cateMap.get(TAGLOAN), appApkObj);
                }
                if (bankTag == 1) {
                    InsertApkToCategory(cateMap.get(TAGBANK), appApkObj);
                }
                if (betTag == 1) {
                    InsertApkToCategory(cateMap.get(TAGBET), appApkObj);
                }

                for (String appTag : appTags) {
                    InsertApkToCategory(cateMap.get(appTag), appApkObj);
                }
            }

            resJson = new JSONObject();
            resJson.put(OUTPUTKEY, user_key);
            resJson.put(CAPTIME, cap_time);
            resJson.put("job_id", job_id);

            totInfo.calcuTotalLabel();

            Set<String> keySet = cateMap.keySet();
            for (String k : keySet) {
                AppCategoryBasicInfo basicInfo = cateMap.get(k);
                String cate = basicInfo.getCate();
                HashMap<String, Object> singleMap = basicInfo.getSingleMap();
                if (!k.equals(TAGTOT)) {
                    if (basicInfo != null) {
                        basicInfo.calcuCateLabel(totInfo.getBasicMap());
                    }
                }
                Set<String> keySets = singleMap.keySet();
                for (String label : keySets) {
                    //                    String suffix = label.equals("") ? "cnt" : label;
                    String suffix = "";
                    if (label.equals("")) {
                        suffix = "cnt";
                    } else if (label.equals("bf_hfq") || label.equals("aft_hfq")) {
                        suffix = label + "_cnt";
                    } else {
                        suffix = label;
                    }
                    resJson.put(cate + "_" + suffix, singleMap.get(label));
                }
            }
            long output_time = System.currentTimeMillis();
            resJson.put(LABELOUTPUTIME, output_time);
        } catch (Exception e) {
            System.out.println(e.toString());
            LOG.error(e.toString());
            LOG.error("PlistRichFlatMapFunc error input: " + inputJson);
            e.printStackTrace();
        }

        return resJson;
    }

    private static void initTagMap() throws IOException {
        File tagFile = new File(APPAPKTAGFILE);
        List<String> lines = FileUtils.readLines(tagFile);
        //        DistributedCache distributedCache = getRuntimeContext().getDistributedCache();
        //        File file = distributedCache.getFile(APPTAGFILENAME);
        //        List<String> lines = FileUtils.readLines(file);
        for (String line : lines) {
            String[] splits = line.split(",");
            if (splits.length == 3) {
                String appName = splits[0].toLowerCase();
                String apkName = splits[1].toLowerCase();
                String tag = splits[2];
                ArrayList<String> appTagArray = appTagMap.get(appName);
                ArrayList<String> apkTagArray = appTagMap.get(apkName);
                if (appTagArray == null) {
                    appTagArray = new ArrayList<>();
                }
                if (apkTagArray == null) {
                    apkTagArray = new ArrayList<>();
                }
                appTagArray.add(tag);
                apkTagArray.add(tag);

                appTagMap.put(appName, appTagArray);
                apkTagMap.put(apkName, apkTagArray);
            }
        }
    }

    private static HashMap<String, AppCategoryBasicInfo> initMap() {
        HashMap<String, AppCategoryBasicInfo> map = new HashMap<>();
        map.put(TAGTOT, new AppCategoryBasicInfo(TAGTOT));
        map.put(TAGBANK, new AppCategoryBasicInfo(TAGBANK));
        map.put(TAGLOAN, new AppCategoryBasicInfo(TAGLOAN));
        map.put(TAGBET, new AppCategoryBasicInfo(TAGBET));
        map.put(TAGJJSH, new AppCategoryBasicInfo(TAGJJSH));
        map.put(TAGJRLC, new AppCategoryBasicInfo(TAGJRLC));
        map.put(TAGLTSJ, new AppCategoryBasicInfo(TAGLTSJ));
        map.put(TAGLYJT, new AppCategoryBasicInfo(TAGLYJT));
        map.put(TAGNONE, new AppCategoryBasicInfo(TAGNONE));
        map.put(TAGSCTY, new AppCategoryBasicInfo(TAGSCTY));
        map.put(TAGSJYX, new AppCategoryBasicInfo(TAGSJYX));
        map.put(TAGSSGW, new AppCategoryBasicInfo(TAGSSGW));
        map.put(TAGSYGJ, new AppCategoryBasicInfo(TAGSYGJ));
        map.put(TAGSYSX, new AppCategoryBasicInfo(TAGSYSX));
        map.put(TAGTSYD, new AppCategoryBasicInfo(TAGTSYD));
        map.put(TAGXLBG, new AppCategoryBasicInfo(TAGXLBG));
        map.put(TAGXWZX, new AppCategoryBasicInfo(TAGXWZX));
        map.put(TAGXXJY, new AppCategoryBasicInfo(TAGXXJY));
        map.put(TAGXXYL, new AppCategoryBasicInfo(TAGXXYL));
        map.put(TAGYLJK, new AppCategoryBasicInfo(TAGYLJK));
        map.put(TAGYYST, new AppCategoryBasicInfo(TAGYYST));
        map.put(TAGOTHER, new AppCategoryBasicInfo(TAGOTHER));
        return map;
    }

    private static void InsertApkToCategory(AppCategoryBasicInfo cateInfo, AppApkObject appApkObj) {
        String cate = cateInfo.getCate();
        cateInfo.addKeySet(AppCategoryBasicInfo.ALLSET, appApkObj.getApp_apk());
        // list
        cateInfo.addInstDay(appApkObj.getInst_day());
        cateInfo.addInstMth(appApkObj.getInst_mth());
        cateInfo.addUpdtDay(appApkObj.getUpdt_day());
        cateInfo.addUpdtMth(appApkObj.getUpdt_mth());
        // set
        cateInfo.addHourPartSet(appApkObj.getApp_apk(), appApkObj.getInst_h_part(), "inst");
        cateInfo.addHourPartSet(appApkObj.getApp_apk(), appApkObj.getUpdt_h_part(), "updt");
        cateInfo.addDayPartSet(appApkObj.getApp_apk(), appApkObj.getInst_day(), "inst");
        //        if(appApkObj.getApp_apk().equals("小米画报|com.mfashiongallery.emag")
        //        || appApkObj.getApp_apk().equals("内容中心|com.miui.newhome")) {
        //            LOG.warn("test");
        //        }
        cateInfo.addDayPartSet(appApkObj.getApp_apk(), appApkObj.getUpdt_day(), "updt");
        cateInfo.addMonPartSet(
                appApkObj.getApp_apk(), appApkObj.getInst_mth(), "inst", cateInfo.getCate());
        cateInfo.addMonPartSet(
                appApkObj.getApp_apk(), appApkObj.getUpdt_mth(), "updt", cateInfo.getCate());

        // 银行和贷款类没有month range
        if (cate.equals(TAGBANK) || cate.equals(TAGLOAN)) {
            if (appApkObj.getInst_hfq_tag() != null) {
                String tmpTag = appApkObj.getInst_hfq_tag();
                if (tmpTag.equals(BFHFQ)) {
                    tmpTag = "bf_hfq";
                } else if (tmpTag.equals(AFTHFQ)) {
                    tmpTag = "aft_hfq";
                }
                cateInfo.addKeySet(tmpTag, appApkObj.getApp_apk());
                /** add patch */
                cateInfo.addInstDayCmpHfqSet(tmpTag, appApkObj);

                cateInfo.addDayPartSet(
                        appApkObj.getApp_apk(),
                        appApkObj.getInst_day(),
                        appApkObj.getInst_hfq_tag() + "_inst");
                cateInfo.addDayPartSet(
                        appApkObj.getApp_apk(),
                        appApkObj.getUpdt_day(),
                        appApkObj.getInst_hfq_tag() + "_updt");
                cateInfo.addMonPartSet(
                        appApkObj.getApp_apk(),
                        appApkObj.getInst_mth(),
                        appApkObj.getInst_hfq_tag() + "_inst",
                        cateInfo.getCate());
                cateInfo.addMonPartSet(
                        appApkObj.getApp_apk(),
                        appApkObj.getUpdt_mth(),
                        appApkObj.getInst_hfq_tag() + "_updt",
                        cateInfo.getCate());
            }
        }
    }

    private static Integer dateBetween(long base_time, Long com_time, SimpleDateFormat sdf)
            throws Exception {
        if (com_time == null || com_time <= INVAILDTIME) {
            //            return CommonTimeTools.dateDiff(captime, INVAILDTIME, sdf);
            return null;
        }

        return CommonTimeTools.dateDiff(base_time, com_time, sdf);
    }

    private static Integer monthBetween(Long base_time, Long com_time) {
        if (com_time == null || com_time <= INVAILDTIME) {
            //            return CommonTimeTools.monthDiff(captime, INVAILDTIME);
            return null;
        }
        return CommonTimeTools.monthDiff(base_time, com_time);
    }

    private static class WeKafkaKeyedSerializationSchema
            implements KeyedSerializationSchema<String> {
        @Override
        public byte[] serializeKey(String element) {
            JSONObject jsonObject = JSONObject.parseObject(element);
            String keyby = jsonObject.getString(OUTPUTKEY);
            return keyby.getBytes();
        }

        @Override
        public byte[] serializeValue(String element) {
            return element.getBytes();
        }

        @Override
        public String getTargetTopic(String element) {
            return null;
        }
    }

    private static class WeKafkaCustomPartitioner extends FlinkKafkaPartitioner<String> {

        @Override
        public int partition(
                String record, byte[] key, byte[] value, String targetTopic, int[] partitions) {
            int partition = Math.abs(new String(key).hashCode() % partitions.length);
            if (LOG.isDebugEnabled()) {
                LOG.info(
                        " partitions: "
                                + partitions.length
                                + " partition: "
                                + partition
                                + " key: "
                                + new String(key));
            }
            return partition;
        }
    }
}
